package org.example.flink.windows;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

public class Hello06TimeWindowAll {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = environment.socketTextStream("localhost", 19523);

        source.map(word -> Tuple2.of(word.split(":")[0],
                        Integer.parseInt(word.split(":")[1])), Types.TUPLE(Types.STRING, Types.INT))
                .windowAll(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(2)))
                .reduce((t1, t2) -> {
                    t1.f1 = t1.f1 + t2.f1;
                    return t1;
                }).map(tuple2 -> {
                    tuple2.f0 =
                            LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy年MM月dd日HH时mm分ss秒SSS毫秒")) + tuple2.f0;
                    return tuple2;
                }, Types.TUPLE(Types.STRING, Types.INT))
                .print("TimeWindow--Sliding:").setParallelism(1);

        environment.execute();
    }
}
